package com.jblv.frame.common.kafka;

import com.jblv.frame.common.kafka.bean.MqBaseMessage;
import com.jblv.frame.common.kafka.bean.ObjectDecoder;
import com.jblv.frame.common.kafka.config.KafkaConfigUtils;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.Decoder;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @author happyonion.yangtao
 * @version 1.0
 * @title KafkaConsumerUtils
 * @projectName jblv-frame
 * @packageName com.jblv.user.service.kafka
 * @description Kafka消息消费工具类
 * @date Create at 2020/4/17 14:05
 * @since jdk1.8
 */

public class KafkaConsumerUtils  {
    private Map<String, ConsumerConnector> consumerMap = new HashMap();
    private static KafkaConsumerUtils utils;

    private KafkaConsumerUtils() {
    }

    public static KafkaConsumerUtils newInstance() {
        if (utils == null) {
            synchronized(KafkaConsumerUtils.class) {
                if (utils == null) {
                    utils = new KafkaConsumerUtils();
                }
            }
        }

        return utils;
    }

    public void close() {
        Iterator var1 = consumerMap.entrySet().iterator();

        while(var1.hasNext()) {
            Map.Entry<String, ConsumerConnector> entry = (Map.Entry)var1.next();
            (entry.getValue()).shutdown();
        }

    }

    public void consume(String[] topics, MessageHandler[] messageHandlers, int[] threads) {
        Decoder objectDecoder = new ObjectDecoder();
        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        int i = 0;
        int count = topics.length;

        for(int offset = 0; offset < count; ++offset) {
            String topic = topics[offset];
            Map<String, Integer> topicCountMap = new HashMap();
            topicCountMap.put(topic, threads[i]);
            ConsumerConnector consumer = Consumer.createJavaConsumerConnector(KafkaConfigUtils.getConsumerConfig());
            consumerMap.put(topic, consumer);
            Map<String, List<KafkaStream<String, MqBaseMessage>>> listMap = consumer.createMessageStreams(topicCountMap, keyDecoder, objectDecoder);
            List<KafkaStream<String, MqBaseMessage>> listStreams = listMap.get(topic);
            int j = 0;
            Iterator var16 = listStreams.iterator();

            while(var16.hasNext()) {
                KafkaStream<String, MqBaseMessage> stream = (KafkaStream)var16.next();
                System.out.println("创建队列 线程： " + topic + " , j = " + j);
                (new Thread(new KafkaConsumerUtils.ConsumeRunnable(stream, messageHandlers[i], topic, j++))).start();
            }

            ++i;
        }

    }

    public void consume(String[] topics, MessageHandler[] messageHandlers) {
        int[] threads = new int[topics.length];

        for(int i = 0; i < topics.length; ++i) {
            threads[i] = 1;
        }

        this.consume(topics, messageHandlers, threads);
    }

    public void consumeStringDecoder(String[] topics, MessageHandler[] messageHandlers, int[] threads) {
        StringDecoder objectDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        int i = 0;
        int count = topics.length;

        for(int offset = 0; offset < count; ++offset) {
            String topic = topics[offset];
            Map<String, Integer> topicCountMap = new HashMap();
            topicCountMap.put(topic, threads[i]);
            ConsumerConnector consumer = Consumer.createJavaConsumerConnector(KafkaConfigUtils.getConsumerConfig());
            this.consumerMap.put(topic, consumer);
            Map<String, List<KafkaStream<String, String>>> listMap = consumer.createMessageStreams(topicCountMap, keyDecoder, objectDecoder);
            List<KafkaStream<String, String>> listStreams = listMap.get(topic);
            int j = 0;
            Iterator var16 = listStreams.iterator();

            while(var16.hasNext()) {
                KafkaStream<String, String> stream = (KafkaStream)var16.next();
                System.out.println("创建队列 线程： " + j + ",队列名称" + topic + "");
                (new Thread(new KafkaConsumerUtils.ConsumeRunnable(stream, messageHandlers[i], topic, j++))).start();
            }

            ++i;
        }

    }

    public void consumeStringDecoder(String[] topics, MessageHandler[] messageHandlers) {
        int[] threads = new int[topics.length];

        for(int i = 0; i < topics.length; ++i) {
            threads[i] = 1;
        }

        this.consumeStringDecoder(topics, messageHandlers, threads);
    }

    class ConsumeRunnable<T> implements Runnable {
        private String topic;
        private int threadNum;
        KafkaStream<String, T> stream = null;
        MessageHandler messageHandler = null;

        @Override
        protected void finalize() throws Throwable {
            super.finalize();
        }

        public String getTopic() {
            return this.topic;
        }

        public void setTopic(String topic) {
            this.topic = topic;
        }

        public ConsumeRunnable(KafkaStream<String, T> stream, MessageHandler messageHandler, String topic, int threadNum) {
            this.stream = stream;
            this.messageHandler = messageHandler;
            this.topic = topic;
            this.threadNum = threadNum;
        }

        @Override
        public void run() {
            ConsumerIterator it = this.stream.iterator();

            while(it.hasNext()) {
                try {
                    this.messageHandler.handler(it.next().message());
                } catch (Exception var3) {
                }
            }

        }
    }
}